package com.atguigu.gmall.realtime.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.DimUtils;
import com.atguigu.gmall.realtime.utils.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeoutException;

/**
 * 维度数据异步关联函数
 *
 * @author lvbingbing
 * @date 2022-06-28 18:54
 */
@Slf4j
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {

    private transient ExecutorService executorService;

    /**
     * 维度表名称
     */
    private String dimTableName;

    private DimAsyncFunction() {
    }

    public DimAsyncFunction(String dimTableName) {
        this.dimTableName = dimTableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 创建线程池对象
        executorService = ThreadPoolUtils.getInstance();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        // 发送异步请求，完成维度关联
        executorService.submit(() -> {
            long start = System.currentTimeMillis();
            String dimKey = null;
            try {
                // 1、获取维度的key
                dimKey = getDimKey(input);
                // 2、根据 dimKey 到维度表中获取维度数据
                JSONObject dimInfoJsonObj = DimUtils.getDimInfo(dimTableName, dimKey);
                // 3、把维度对象的属性赋值给流中的对象属性
                if (MapUtils.isEmpty(dimInfoJsonObj)) {
                    String s = JSON.toJSONString(input);
                    log.warn("维度异步查询未获取到数据！要进行维度关联的事实数据：{}, 维度表：{}, 维度key：{}", s, dimTableName, dimKey);
                } else {
                    setDimField(input, dimInfoJsonObj);
                }
                long end = System.currentTimeMillis();
                long costTime = end - start;
                log.info("维度异步查询耗时：{} 毫秒。维度表：{}, 维度key：{}", costTime, dimTableName, dimKey);
                // 4、维度关联后的结果
                resultFuture.complete(Collections.singleton(input));
            } catch (Exception e) {
                String s = JSON.toJSONString(input);
                log.error("维度异步查询出现异常！要进行维度关联的事实数据：{}, 维度表：{}, 维度key：{}, 异常原因：{}", s, dimTableName, dimKey, e);
            }
        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        log.error("维度数据异步查询超时！要进行维度关联的事实数据：{}, 维度表：{}", input, dimTableName);
        resultFuture.completeExceptionally(
                new TimeoutException("维度数据异步查询超时！"));
    }
}